/*
 * @Author: your name
 * @Date: 2020-05-12 17:05:16
 * @LastEditTime: 2020-09-02 15:46:12
 * @LastEditors: Please set LastEditors
 * @Description: In User Settings Edit
 * @FilePath: \stroage_server\platform\av_play\av_play.c
 */

#include "global_common.h"
#include "av_play.h"
#include "xeSqlite.h"
#include "file_space.h"
#include "read_ps_file.h"
#include "my_timer.h"

#include "list.h"

#include "userMgntApi.h"
typedef enum __av_opt_type{
    E_OPT_UNKNOW = 0,
    E_OPT_DOWNLOAD = 1,
    E_OPT_AVPLAY = 2,
}e_av_opt_type;

typedef enum __file_pos{
    E_FILE_UNKNOW = 0,
    E_FILE_FIRST = 0x1,
    E_FILE_MIDDLE = 0x2,
    E_FILE_LAST = 0x4,
}e_file_pos;

typedef struct FILE_TRANSMIT {
    int session_id;
	char name[HISOME_LEN_64];
	int listen_sockfd, send_sockfd;
    e_av_opt_type av_opt_type;
	int proto;
	int mode;

	struct sockaddr_in addr_peer;
	struct sockaddr_in addr_local;

	void * priv;

    FILE *pFd;

    t_ipc_query_info ipc_info;
    t_ipc_query_result ipc_query_result;

    struct list_head list;
    float  speed;
    int exit_num;

    long long total_size;
    long long send_size;
    t_session play_session;
} file_play_t;

typedef struct _playInfo{
    char          sid[HISOME_LEN_128];  
    char          ipc_SN[HISOME_LEN_128];
    char          ipc_MAC[HISOME_LEN_128];

    char          opt_type[HISOME_LEN_16];
    char          opt_mode[HISOME_LEN_16];
      
    char          filename[HISOME_LEN_128];

    char          _startTime[HISOME_LEN_128];
    char          _endTime[HISOME_LEN_128];

    unsigned int  rec_listenPort;//srv端监听的tcp端口
    int           session_id;

    long long     offset;
    long long     file_size;

    char          mode[HISOME_LEN_16];
    char          proto[HISOME_LEN_16];
}t_playInfo;

LIST_HEAD(g_play_list);

static int ft_task_thread_num = 0;

static int listen_port_base=19000;
static int listen_port_end=19100;
static int listen_port_current=19000;

static int __ft_task_TCP_listen(int sockfd) {
	struct sockaddr_in addr;
    int ret;

	addr.sin_family=AF_INET;
	addr.sin_addr.s_addr=INADDR_ANY;

	listen_port_current++;
    {
        int recvbuflen = (1<<20);
        if(setsockopt(sockfd,SOL_SOCKET,SO_SNDBUF,(void*)&recvbuflen,sizeof(int))< 0){
            recvbuflen = (1<<20);
            setsockopt(sockfd,SOL_SOCKET,SO_SNDBUFFORCE,(void*)&recvbuflen,sizeof(int));
        }
    }
	while ( 1) {
		if (listen_port_current>listen_port_end)
			listen_port_current=listen_port_base;

		addr.sin_port=htons(listen_port_current);
		if (bind(sockfd, (struct sockaddr*)&addr, sizeof(addr))==0)
			break;

		xeLOG_ERR("bind(%d) FAIL, %s",listen_port_current,strerror(errno));

		listen_port_current++;
	}
	ret = listen(sockfd, 5);
    if(ret < 0)
        return -1;
	return listen_port_current;
}

static int __ft_task_set_TCP_keepalive(int sockfd){
	/**
	* Aug-20-2008 by dengy<dengy@hisome.com>
	* add TCP keepalive;
	**/
	int keepAlive = 1;
	int keepIdle = 6;
	int keepInterval = 1;
	int keepCount = 3;

	/**
	* SO_KEEPALIVE
	*     Enable sending of keep-alive messages on connection-oriented sockets. 
	*     Expects a integer boolean flag.  
	**/
	setsockopt(sockfd,SOL_SOCKET,SO_KEEPALIVE, (void*)&keepAlive,sizeof(keepAlive));
	/**
	* TCP_KEEPIDLE
	*     The time (in seconds) the connection needs to remain idle before TCP 
	*     starts  sending  keepalive probes, if the socket option SO_KEEPALIVE 
	*     has been set on this socket.  This option should not be used in code 
	*     intended to be portable.  
	**/
	setsockopt(sockfd,SOL_TCP, TCP_KEEPIDLE,(void *)&keepIdle,sizeof(keepIdle));
	/**
	* TCP_KEEPINTVL
	*     The time (in seconds) between individual keepalive probes. This option
	*     should not be used in code intended to be portable.
	**/
	setsockopt(sockfd,SOL_TCP,TCP_KEEPINTVL,(void *)&keepInterval,sizeof(keepInterval));
	/**
	* TCP_KEEPCNT
	*	  The maximum number of keepalive probes TCP should send before dropping 
	*	  the connection. This option should not be used in code intended to be portable.  
	**/
	setsockopt(sockfd,SOL_TCP, TCP_KEEPCNT,(void *)&keepCount,sizeof(keepCount));

	return 0;
}

static int _ft_task_set_address_TCP_PASSIVE(file_play_t *ftp, char *address, size_t addr_len) {
	int port=0;
	// char lip[64] = {0};
	// int IP0,IP1,IP2,IP3;
	struct sockaddr_in *saip=NULL;
	int sockfd;
    ftp->listen_sockfd =-1;
    ftp->send_sockfd = -1;
	sockfd = socket(AF_INET, SOCK_STREAM, 0);
	if (sockfd == -1) {
		xeLOG_ERR("socket FAIL, %s",strerror(errno));
		return -1;
	}

	__ft_task_set_TCP_keepalive(sockfd);
	port = __ft_task_TCP_listen(sockfd);

	ftp->listen_sockfd = sockfd;

	//filled the addr_local;
	saip=(struct sockaddr_in*)&ftp->addr_local;
	saip->sin_family      = AF_INET;
	saip->sin_addr.s_addr = INADDR_ANY;//address
	saip->sin_port        = htons(port);

	ftp->proto	= FT_TASK_PROTO_TCP;
	ftp->mode	= FT_TASK_MODE_PASSIVE;
	return 0;
}
static int _ft_task_startup(file_play_t *ftp){
	int ret;
	socklen_t addr_len;
	int new_sockfd = -1;
	fd_set rfds;
	char peerip[64] = {0};
	struct timeval tout;


	if (ftp->proto == FT_TASK_PROTO_TCP &&
		ftp->mode == FT_TASK_MODE_PASSIVE) 
	{
		FD_ZERO(&rfds);
		FD_SET(ftp->listen_sockfd, &rfds);
		tout.tv_sec  = 10;
		tout.tv_usec = 0;

		ret = select(ftp->listen_sockfd+1, &rfds, NULL, NULL, &tout);
		if(ret == -1){
			xeLOG_INFO("%s select error, %s", ftp->name, strerror(errno));
			goto play_startup_out;
		}

		if ( !FD_ISSET(ftp->listen_sockfd, &rfds) ) {
			xeLOG_INFO("%s select timeout port:%d", ftp->name,(htons)(ftp->addr_local.sin_port));
			goto play_startup_out;
		}

		addr_len = sizeof(struct sockaddr_in);
		new_sockfd = accept(ftp->listen_sockfd, (struct sockaddr*)&ftp->addr_peer, &addr_len);
		if (new_sockfd == -1) {
			xeLOG_ERR("%s accept FAIL, %s", ftp->name,strerror(errno));
			goto play_startup_out;
		}

		/*close the listen socket*/
		close(ftp->listen_sockfd);
		ftp->listen_sockfd = -1;

		ftp->send_sockfd = new_sockfd;
		addr_len = sizeof(struct sockaddr_in);
		ret = getpeername(ftp->send_sockfd, (struct sockaddr*)&ftp->addr_peer, &addr_len);
		if (ret==-1) {
			xeLOG_ERR("%s getpeername FAIL, %s",ftp->name,strerror(errno));
			goto play_startup_out;
		}

		inet_ntop(AF_INET, &ftp->addr_peer.sin_addr, peerip, sizeof(peerip)); 
		snprintf( ftp->name, HISOME_LEN_64-1, "ch_%s_%d", 
				peerip, ntohs(ftp->addr_peer.sin_port));


        int on=1;
        ret = ioctl(ftp->send_sockfd, FIONBIO, &on);
        if(ret == -1){
            xeLOG_NOTICE("ioctl(FIONBIO) fail, %s",strerror(errno));
            return -1;
        }

        setsockopt(ftp->send_sockfd, IPPROTO_TCP, TCP_NODELAY,(void *)&on, sizeof(int));

		int no = 1;
        int recvbuflen = (1<<20);
		setsockopt(ftp->send_sockfd, IPPROTO_TCP, TCP_CORK, (char*)&no, sizeof(int));
    
        if(setsockopt(ftp->send_sockfd,SOL_SOCKET,SO_SNDBUF,(void*)&recvbuflen,sizeof(int))< 0){
            recvbuflen = (1<<20);
            setsockopt(ftp->send_sockfd,SOL_SOCKET,SO_SNDBUFFORCE,(void*)&recvbuflen,sizeof(int));
        }

        struct linger lon;
        lon.l_onoff = 0;
        lon.l_linger = 0;
        setsockopt(ftp->send_sockfd, SOL_SOCKET, SO_LINGER, &lon,sizeof(struct linger));
    
        __ft_task_set_TCP_keepalive(ftp->send_sockfd);

	}
	return 0;  

play_startup_out:
	if(ftp->listen_sockfd != -1)
		close(ftp->listen_sockfd);
	if(ftp->send_sockfd != -1)
		close(ftp->send_sockfd);

	ftp->listen_sockfd = -1;
	ftp->send_sockfd   = -1;

	return -1;
}
#define DOWN_LOAD_LEN ((1<<10)*250)
int proc_download(file_play_t *ftp,t_file_opt_info *opt_info)
{
    int ret;
    int each_send_len = 0;
    long long total_left_bytes;
    int send_tmp=0;
    char data_buf[DOWN_LOAD_LEN+200]={0};

    total_left_bytes = opt_info->end_offset_bytes - opt_info->start_offset_bytes;
    STD_print("donload,%s,%lld",opt_info->file_des.file_name,total_left_bytes);
    ftp->pFd = fopen(opt_info->file_des.file_name,"rb");
    if(!ftp->pFd){
        xeLOG_ERR("open FAIL,%s",opt_info->file_des.file_name);
         ftp->send_size +=total_left_bytes;
        return -1;
    }
    fseek(ftp->pFd, opt_info->start_offset_bytes,SEEK_SET);

    while(20200513){
        int read_len;
        if (!feof (ftp->pFd)){
            if(total_left_bytes == 0){
                STD_print("send this file over,total_left_bytes is 0 %s",opt_info->file_des.file_name);
                break;
            }
            if(total_left_bytes >DOWN_LOAD_LEN ){
                each_send_len = DOWN_LOAD_LEN;
            }else{
                each_send_len = total_left_bytes;
            }
            read_len = fread(data_buf,1,each_send_len,ftp->pFd);
            if(read_len != each_send_len){
                STD_print("warning,fread failed read_len:%d,each_send_len:%d",read_len,each_send_len);
            }
            send_tmp = 0;
send_again:
            ret = send(ftp->send_sockfd,data_buf+send_tmp,read_len-send_tmp,MSG_NOSIGNAL);
            if(ret < 0){
                if(ret ==-1 && (errno==EAGAIN || errno==EWOULDBLOCK)){
                    // STD_print("would block");
                    usleep(100*1000);
                    goto send_again;
                }else if(ret  < 0){
                    STD_print("send failed ret:%d errno:%d,%s",
                        ret,errno,strerror(errno));
                    break;
                }
            }else{
                ftp->send_size += ret;
                total_left_bytes -= ret;
                // {
                //     static long long bytes_send= 0;
                //     bytes_send += ret;
                //     STD_print("send :%d total send:%lld,left:%lld",ret,bytes_send,total_left_bytes);
                // }
                if(ret != (read_len-send_tmp)){
                    // STD_print("dl warning:send:%d, read:%d",ret,read_len-send_tmp);
                    send_tmp += ret;
                    goto send_again;
                }
                

                usleep(25*1000);
            }
        }else{
            STD_print("read file end total_left_bytes:%lld",total_left_bytes);
            break;
        }
    }
    if(ftp->pFd){
        fclose(ftp->pFd);
        ftp->pFd = NULL;
    }
    return 0;
}
int proc_avplay(file_play_t *ftp,t_file_opt_info *opt_info)
{
    int ret;

    uint32_t ts =0;
    PS_t ps_unit={0};
    int send_tmp=0;

    long long left_bytes;

    ret = ps_read_init(&ps_unit,opt_info->file_des.file_name);
    if(ret < 0){
        xeLOG_ERR("ps_read_init failed:%s",opt_info->file_des.file_name);
        return -1;
    }
    left_bytes = opt_info->end_offset_bytes - opt_info->start_offset_bytes;
    fseek(ps_unit.psbitstream, opt_info->start_offset_bytes,SEEK_SET);

    user_session_init(&ftp->play_session);

    while(20200514){
        char *p_buf;
        int read_len;

        ts += (uint32_t)(ftp->speed * 40);
        user_session_send_with_ts(&ftp->play_session,ts);//wait for next 40ms
        if(ftp->exit_num == S_AV_PLAY_PAUSE){
            continue;
        }else if(ftp->exit_num == S_AV_PLAY_STOP){
            break;
        }
        ret = ps_read_one_frame(&ps_unit,&p_buf,&read_len);
        if(ret < 0 ){
            break;
        }

play_again:
        ret = send(ftp->send_sockfd,p_buf+send_tmp,read_len-send_tmp,MSG_NOSIGNAL);
        if(ret < 0){
            if(ret ==-1 && (errno==EAGAIN || errno==EWOULDBLOCK)){
                STD_print("would block ftp->send_size:%lld",ftp->send_size);
                usleep(40*1000);
                goto play_again;
            }else if(ret  < 0){
                xeLOG_WARNING("send failed ret:%d errno:%d,%s,ftp->send_size:%lld",
                    ret,errno,strerror(errno),ftp->send_size);
                break;
            }
        }else{
            ftp->send_size += ret;
            left_bytes -= ret;
            if(ret != (read_len-send_tmp)){
                // STD_print("dl warning:send:%d, read:%d",ret,read_len-send_tmp);
                send_tmp += ret;
                goto play_again;
            }
            if(left_bytes <= 0){
                break;
            }
        }
        // xeLOG_INFO("read_len:%d send_len:%d",read_len,ret);
    }
    user_session_deinit(&ftp->play_session);
    ps_read_deinit(&ps_unit);
    xeLOG_INFO("%s end,left_bytes:%lld",opt_info->file_des.file_name,left_bytes);
    return 0;
}
int play_proc_data_2(file_play_t *ftp)
{
    int ret;
    vector_iter iter;
    vector_t *c = ftp->ipc_query_result.p_vector;
    c->tmp_cursor = 0;
    xeLOG_NOTICE("start proc port:%d for %s,vector size:%ld ",(htons)(ftp->addr_local.sin_port),
        (ftp->av_opt_type == E_OPT_DOWNLOAD)?"download":"avplay",c->size);

    for (iter = vector_begin(c); iter != vector_end(c); iter = vector_next(c)) {
        t_file_opt_info *tt = vector_iter_valuep(c, iter, t_file_opt_info);
        if(ftp->av_opt_type == E_OPT_DOWNLOAD){
            ret += proc_download(ftp,tt);
        }else{
            ret += proc_avplay(ftp,tt);
        }
    }
    return ret;
}

void * download_play_workthread(void * arg){
    // int ret = 0;
    char thread_name[128]="";
	file_play_t *ftp = (file_play_t *)arg;

    int port = (htons)(ftp->addr_local.sin_port);
	pthread_detach(pthread_self());
    sprintf(thread_name,"%s",ftp->ipc_info.ipc_sn);
    prctl(PR_SET_NAME, thread_name);

    ft_task_thread_num++;
	if(-1 == _ft_task_startup(ftp) ) { 
		goto play_task_out;
	}

    play_proc_data_2(ftp);
play_task_out:
    if(ftp){
        if(ftp->send_sockfd != -1) {
            STD_print("close send_sockfd");
            close(ftp->send_sockfd);
            ftp->send_sockfd = -1;
        }
        if(ftp->pFd){
            fclose(ftp->pFd);
            ftp->pFd = NULL;
            sync();
        }
        vector_destroy(ftp->ipc_query_result.p_vector);
        ftp->ipc_query_result.p_vector =NULL;
        if(ftp->session_id)
            list_del(&ftp->list);
    }
    __free(ftp);
    ft_task_thread_num--;

	xeLOG_NOTICE("ft_task_thread(%ld): localport:%d exit!! task_num : %d", _CORE_GETTID(), port, ft_task_thread_num);
    pthread_exit(0);
}



int play_http_respond(MyHttpPack* req,int get_fail,char *errReason,  t_playInfo *playinfo)
{
    cJSON *rsp,*rspHeader,*rspArr,*rspData;
    rsp=cJSON_CreateArray();
    rspHeader = cJSON_CreateObject();
    rspData = cJSON_CreateObject();
    cJSON_AddStringToObject(rspHeader,"method","POST");
    cJSON_AddStringToObject(rspHeader,"SID", playinfo->sid);
    cJSON_AddStringToObject(rspHeader,"SN", playinfo->ipc_SN);
    cJSON_AddStringToObject(rspHeader,"startTime",playinfo->_startTime);
    cJSON_AddStringToObject(rspHeader,"endTime",playinfo->_endTime);
    cJSON_AddItemToObject(rspHeader, "data", rspArr=cJSON_CreateArray());
    cJSON_AddItemToArray(rsp,rspHeader);
    cJSON_AddItemToArray(rspArr,rspData);
    if(get_fail){
        cJSON_AddStringToObject(rspHeader,"result","FAIL");

        cJSON_AddStringToObject(rspData,"reason",errReason);
    }else{
        cJSON_AddStringToObject(rspHeader,"result","SUCCESS");

        cJSON_AddNumberToObject(rspData,"port",playinfo->rec_listenPort);
        cJSON_AddNumberToObject(rspData,"sessionid",playinfo->session_id);
    }
    com_httpPack(req->httpReq.nc,rsp);
    cJSON_Delete(rspHeader);

    return 0;
}


void dowdload_play_setup(void *data)
{
    pthread_t thread_id;
    int ret;
    int ret_flag = 0;
    char errReason[HISOME_LEN_512]={'\0'};
    t_playInfo playinfo ={0};
    MyHttpPack* req =(MyHttpPack*)data; 
    cJSON *root;
    file_play_t *ftp = NULL;
    // int session_id = 0;

    root=cJSON_Parse(req->httpReq.hm->body.p);
    if (!root) {
        sprintf(errReason,"客户端发送数据有误");
        ret_flag=1;
        PRTLOG_ERR("Error before: [%s]",cJSON_GetErrorPtr());
        goto play_exit;
    }else{
        char value[HISOME_LEN_256];
        
        int userid = 0;
        __prase_cjson_string(value,ret_flag,"userID",play_exit);
        sscanf(value,"%d",&userid);
        if(NULL ==apiUsrMgnt_findOn_LINEUser(userid)){
            xeLOG_ERR("error, %d not online",userid);
            ret_flag = 1;
            sprintf(errReason,"error, %d not online",userid);
            goto play_exit;
        }

        __prase_cjson_string(value,ret_flag,"SID",play_exit);
            strcpy(playinfo.sid,value);
        __prase_cjson_string(value,ret_flag,"SN",play_exit);
            strcpy(playinfo.ipc_SN,value);
        __prase_cjson_string(value,ret_flag,"opt_type",play_exit);
            strcpy(playinfo.opt_type,value);
        __prase_cjson_string(value,ret_flag,"opt_mode",play_exit);
            strcpy(playinfo.opt_mode,value);
	    if (!strcmp(playinfo.opt_mode, "file")){
            // __prase_cjson_string(value,ret_flag,"filename",play_exit);
            //     strcpy(playinfo.filename,value);
            // __prase_cjson_string(value,ret_flag,"offset",play_exit);
            //     sscanf(value,"%lld",&playinfo.offset);
        }else if (!strcmp(playinfo.opt_mode, "time")) {
            __prase_cjson_string(value,ret_flag,"startTime",play_exit);
                strcpy(playinfo._startTime,value);
            __prase_cjson_string(value,ret_flag,"endTime",play_exit);
                strcpy(playinfo._endTime,value);
        }

        __prase_cjson_string(value,ret_flag,"mode",play_exit);
            strcpy(playinfo.mode,value);
    }
    
	ftp = (file_play_t *)calloc(1, sizeof(file_play_t));
    __check_return_point(ftp,ret_flag,"file_play_t calloc failed",play_exit);
    /*
     *1 query ipc_sn in sysmgnt.db
     */
    t_ipc_query_info *ipc_info = &ftp->ipc_info;
    memset(ipc_info, 0 ,sizeof(t_ipc_query_info));
    strcpy(ipc_info->ipc_sn, playinfo.ipc_SN);
    ipc_info->start_time =  com_localTimeStr2Int(playinfo._startTime);
    ipc_info->end_time =  com_localTimeStr2Int(playinfo._endTime);

    xeLOG_DEBUG("http request\n"    \
        "\tipc_info->ipc_sn     : %s\n"    \
        "\tipc_info->ipc_mac    : %s\n"    \
        "\tipc_info->file_len   : %lld\n"    \
        "\tipc_info->start_time  : %ld\n"    \
        "\tipc_info->end_time    : %ld\n\n",
        ipc_info->ipc_sn,
        ipc_info->ipc_mac,
        ipc_info->file_len,
        ipc_info->start_time,
        ipc_info->end_time);
    tiny_rec_query_sysmgntdb_info(E_TBL_IPCMGNT,(void *)ipc_info);
    __check_return_cstring(ipc_info->storage_path,ret_flag,"this sn record in ipc.db",play_exit);

    {
        t_ipc_query_result *p_query_result = &ftp->ipc_query_result;
        memset(p_query_result, 0 ,sizeof(t_ipc_query_result));
        p_query_result->p_vector = vector_create(t_file_opt_info);
        ret = get_reccords_from_ipctable(E_QUERY_BY_TIME,ipc_info,p_query_result);
        if(p_query_result->durNum == 0){
            xeLOG_ERR("no record in %s/ipc.db ",ipc_info->storage_path);
            ret_flag =1;
            sprintf(errReason,"no records found");
            goto play_exit;
        }else{

            vector_iter iter;
            vector_t *c = p_query_result->p_vector;
            for (iter = vector_begin(c); iter != vector_end(c); iter = vector_next(c)) {
                t_file_opt_info *tt = vector_iter_valuep(c, iter, t_file_opt_info);
                tt->start_offset_bytes = get_offset_by_sec(tt->file_des.file_name,
                    tt->file_des.detail_info_name,
                    tt->start_offset_sec,1);
                tt->end_offset_bytes = get_offset_by_sec(tt->file_des.file_name,
                    tt->file_des.detail_info_name,
                    tt->end_offset_sec,0);
                ftp->total_size += (tt->end_offset_bytes - tt->start_offset_bytes);
                STD_print("\n\nvector-->file_name: %s dura:%d start play:%d end play:%d %lld-%lld\n",
                     tt->file_des.file_name,
                     tt->file_des.duration,
                     tt->start_offset_sec,
                     tt->end_offset_sec,
                     tt->start_offset_bytes,
                     tt->end_offset_bytes);
            }
        }
    }
    if (!strcmp(playinfo.opt_type, "download")){
        ftp->av_opt_type = E_OPT_DOWNLOAD;
    }else if (!strcmp(playinfo.opt_type, "avplay")) {
        ftp->av_opt_type = E_OPT_AVPLAY;
        ftp->speed = 1.0;
    }else{
        xeLOG_WARNING("");
    }

    tiny_get_unique_id(&playinfo.session_id);
    ftp->session_id = playinfo.session_id;

    INIT_LIST_HEAD(&ftp->list);
    list_add_tail(&ftp->list, &g_play_list);

	if ( !strcmp(playinfo.mode, "PASSIVE") ) {
        char addr[32]="";
		ret = _ft_task_set_address_TCP_PASSIVE(ftp, addr, sizeof(addr));
		__check_return_val(ret, ret_flag,"listen failed",play_exit);
	} else if ( !strcmp(playinfo.mode, "PORT")) {
        
    }
    playinfo.rec_listenPort = (htons)(ftp->addr_local.sin_port);
    xeLOG_DEBUG("play thread startftp->port:%d",playinfo.rec_listenPort);

	ret = pthread_create(&thread_id, NULL, download_play_workthread, (void*)ftp);
	__check_return_val(ret, ret_flag,"create thread fail",play_exit);

play_exit:
    if(ret_flag){
        if(ftp){
            if(ftp->send_sockfd != -1) {
                close(ftp->send_sockfd);
                ftp->send_sockfd = -1;
            }
            if(ftp->send_sockfd != -1) {
                close(ftp->send_sockfd);
                ftp->send_sockfd = -1;
            }
            if(ftp->pFd){
                fclose(ftp->pFd);
                ftp->pFd = NULL;
                sync();
            }
            vector_destroy(ftp->ipc_query_result.p_vector);
            ftp->ipc_query_result.p_vector =NULL;
            if(ftp->session_id)
                list_del(&ftp->list);
        }
        __free(ftp);
    }
    if(root)
        cJSON_Delete(root);
    play_http_respond(req,ret_flag,errReason,&playinfo);

}

file_play_t * get_session_by_id(int session_id)
{

	struct list_head *entry;
	file_play_t *ftp;
	list_for_each(entry, &g_play_list){
		ftp = list_entry(entry, file_play_t, list);
		if(NULL == ftp){
			break;
		}
		if(ftp->session_id == session_id){
            xeLOG_NOTICE("find ftp");
            return ftp;
		}
	}
    return NULL;
}


void av_ctl_setup(void *data)
{
    cJSON *root;
    int ret_flag = 0;
    char errReason[HISOME_LEN_512]={'\0'};
    MyHttpPack* req =(MyHttpPack*)data;
    float speed=1.0;
    int opttype;
    int rate;
    int session_id;
    file_play_t *ftp = NULL;

    root=cJSON_Parse(req->httpReq.hm->body.p);
    if (!root) {
        sprintf(errReason,"客户端发送数据有误");
        ret_flag=1;
        PRTLOG_ERR("Error before: [%s]",cJSON_GetErrorPtr());
        goto ctrl_exit;
    }else{
        char value[HISOME_LEN_256];
        int userid = 0;
        __prase_cjson_string(value,ret_flag,"userID",ctrl_exit);
        sscanf(value,"%d",&userid);
        if(NULL ==apiUsrMgnt_findOn_LINEUser(userid)){
            xeLOG_ERR("error, %d not online",userid);
            ret_flag = 1;
            sprintf(errReason,"error, %d not online",userid);
            goto ctrl_exit;
        }
        __prase_cjson_string(value,ret_flag,"sessionid",ctrl_exit);
        sscanf(value,"%d",&session_id);
        __prase_cjson_string(value,ret_flag,"opttype",ctrl_exit);
        sscanf(value,"%d",&opttype);

        __prase_cjson_string(value,ret_flag,"speed",ctrl_exit);
            sscanf(value,"%f",&speed);

    }
    ftp = get_session_by_id(session_id);
    __check_return_point(ftp, ret_flag, "can not find thread by session_id",ctrl_exit);
    if(opttype == HTTP_REQ_SPEED_CTRL){
        STD_print("speed:%f",speed);
        ftp->speed = (1.0/speed);
    }else if(opttype == HTTP_REQ_AVPLAY_PAUSE){
        ftp->exit_num = S_AV_PLAY_PAUSE;
    }else if(opttype == HTTP_REQ_AVPLAY_RESUME){
        ftp->exit_num = S_AV_PLAY_RESUME; 
    }else if(opttype == HTTP_REQ_AVPLAY_STOP){
        ftp->exit_num = S_AV_PLAY_STOP; 
    }else if(opttype == HTTP_REQ_AV_DL_RATE){
        rate = (int)(((double)ftp->send_size/(double)ftp->total_size)*100);
    }else{
        
    }
ctrl_exit:
    {
        cJSON *rsp,*rspHeader,*rspArr,*rspData;

        rsp=cJSON_CreateArray();
        rspHeader = cJSON_CreateObject();
        rspData = cJSON_CreateObject();

        cJSON_AddStringToObject(rspHeader,"method","POST");
        cJSON_AddItemToObject(rspHeader, "data", rspArr=cJSON_CreateArray());

        if(ret_flag){
            cJSON_AddStringToObject(rspHeader,"result","FAIL");
            cJSON_AddStringToObject(rspData,"reason",errReason);
            cJSON_AddNumberToObject(rspData,"sessionid",session_id);
        }else{
            cJSON_AddStringToObject(rspHeader,"result","SUCCESS");
            if(opttype == HTTP_REQ_AV_DL_RATE){
                cJSON_AddNumberToObject(rspData,"rate",rate);
                cJSON_AddNumberToObject(rspData,"sessionid",session_id);
            }
        }
        
        cJSON_AddItemToArray(rsp,rspHeader);
        cJSON_AddItemToArray(rspArr,rspData);

        com_httpPack(req->httpReq.nc,rsp);
        cJSON_Delete(rsp);  
    }
    if(root)
        cJSON_Delete(root);
}